/*
 *  Arrow
 *
 *  Created by David Giovannini on 1/5/07.
 *  Copyright 2007 Software by Jove. All rights reserved.
 *
 */
 
#include "SBJ/Arrow/MessageHolder.h"
#include "SBJ/Arrow/Subscription.h"
#include "SBJ/Arrow/MsgProxy.h"

#include <vector>

namespace SBJ
{

namespace Arrow
{

MessageHolder::MessageHolder( int numThreads, const IExceptionObserver::Ptr& excObserver )
: m_threadPool( numThreads )
, m_started( false )
{
}
	
MessageHolder::~MessageHolder( void )
{
}
	
bool
MessageHolder::start( void )
{
    if ( m_started == false )
	{
		m_started = true;
		pushOutSubmittable();
	}
	return ( m_started == true );
}

bool
MessageHolder::stop( bool join )
{
	if ( m_started == true )
	{
		m_started = false;
	}
	return ( m_started ==  false );
}

bool
MessageHolder::addRouter( const IRouter::Ptr& router )
{
	m_router = router;
	pushOutSubmittable();
	return true;
}

bool
MessageHolder::removeRouter( const IRouter::Ptr& router )
{
	if ( m_router == router )
	{
		m_router.reset();
		return true;
	}
	return false;
}

void
MessageHolder::pushOutSubmittable( void )
{
	if ( m_router && m_started )
	{
		std::vector< MsgQueue* > toSubmit;
		toSubmit.reserve( m_messages.size() );
		{
			boost::mutex::scoped_lock lock( m_guard );
			KeyedLists::iterator iter = m_messages.begin();
			while ( iter != m_messages.end() )
			{
				int& refCount = iter->second.first;
				MsgQueue*& temp = iter->second.second;
				if ( refCount > 0 && temp != 0 && temp->size() > 0 )
				{
					toSubmit.push_back( temp );
					temp = 0;
				}
			}
		}
		for ( size_t i = 0; i < toSubmit.size(); ++i )
		{
			m_stats.accumeRelease();
			Submit submit( toSubmit[ i ], *m_router );
			m_threadPool.addTask( submit );
		}
	}
}
	
void
MessageHolder::addInterest( const Subscription& subscription )
{
	MsgQueue* queue = 0;
	{
		boost::mutex::scoped_lock lock( m_guard );
		KeyedLists::iterator findIt = m_messages.find( subscription.getTopic() );
		if ( findIt != m_messages.end() )
		{
			int& refCount = findIt->second.first;
			MsgQueue*& temp = findIt->second.second;
			
			++refCount;
			if ( m_router && m_started && refCount == 1 && temp != 0 && temp->size() > 0 )
			{
				queue = temp;
				temp = 0;
			}
		}
	}
	// use task dispatcher
	if ( queue != 0 )
	{
		m_stats.accumeRelease();
		Submit submit( queue, *m_router );
		m_threadPool.addTask( submit );
	}
}

void
MessageHolder::Submit::operator () ( void )
{
	Message* msg = 0;
	while ( ( msg = m_queue->pop_front() ) != 0 )
	{
		m_router.onMessage( *msg );
		delete msg;
	}
	delete m_queue;
}
	
void
MessageHolder::removeInterest( const Subscription& subscription )
{
	boost::mutex::scoped_lock lock( m_guard );
	KeyedLists::iterator findIt = m_messages.find( subscription.getTopic() );
	if ( findIt != m_messages.end() )
	{
		--findIt->second.first;
		if ( findIt->second.second == 0 && findIt->second.first == 0 )
		{
			findIt->second.second = new MsgQueue;
		}
	}
}
	
void
MessageHolder::onMessage( Message& msg )
{
	// Is there a race condition between this check and addInterest?
	// Feeds are not informed until after the holder, good.
	if ( msg.getTopic().hasSubscription() == false )
	{
		m_stats.accumeHold();
		boost::mutex::scoped_lock lock( m_guard );
		const Topic& topic = msg.getTopic();
		std::pair< KeyedLists::iterator, bool > insertIt = m_messages.insert( std::make_pair( topic, std::make_pair( 0, (MsgQueue*)0 ) ) );
		if ( insertIt.first->second.second == 0 )
		{
			insertIt.first->second.second = new MsgQueue;
		}
		insertIt.first->second.second->push_back( topic.getMsgProxy().duplicate( msg ) );
	}
	else if ( m_router ) // has subscriber or executable
	{
		m_stats.accumePass();
		m_router->onMessage( msg );
	}
}

}

}
